package org.apm.data.client;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apm.data.bean.ScrollBean;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.SearchHit;

public class ScrollProcessor {
    private static final Logger LOGGER = LogManager.getLogger(ScrollProcessor.class);

    private ScrollProcessor() {}

    public static void process(final ScrollBean scrollBean) {
        try {
            Client client = scrollBean.getClient();
            String index = scrollBean.getIndex();

            SearchResponse scrollResp = client.prepareSearch(index).setScroll(new TimeValue(scrollBean.getScroll_time()))
                    .setSize(scrollBean.getScroll_size()).setQuery(scrollBean.getQuery()).execute().actionGet();
            long totalHits = scrollResp.getHits().getTotalHits();
            LOGGER.info("[Analyzer-After][" + index + "] " + totalHits + ".");

            long processSize = 0;
            while (true) {

                // 当请求返回的Hits列表大小为0时,即数据处理完毕.
                if (scrollResp.getHits().getHits().length == 0) {
                    LOGGER.info("[Analyzer-After][" + index + "] total: " + totalHits + ", process: " + processSize);
                    break;
                } else {
                    LOGGER.info("[Analyzer-After][" + index + "] total: " + totalHits + ", process: " + processSize);
                    processSize += scrollResp.getHits().getHits().length;
                }

                for (SearchHit hit : scrollResp.getHits()) {
                    scrollBean.processData(hit);
                }

                scrollResp = client.prepareSearchScroll(scrollResp.getScrollId())
                        .setScroll(new TimeValue(scrollBean.getScroll_time())).execute().actionGet();
            }
        } catch (Exception exception) {
            LOGGER.info(exception.getMessage());
        }
    }

}
